//
// Created by root on 11/7/18.
//

#include <cstdlib>
#include <string.h>
#include <errno.h>
#include <limits.h>
#include <time.h>

#include "st.h"

//#define USE_LOCAL_LOG_LEVEL
#include "ursax.h"
#include "log.h"
#include "utcache.h"
#include "st-redis.h"
#include "networking.h"
#include "redis-single.h"
//#include "redis-cluster.h"
#include "redis-client.h"
#include "endpoints.h"


char chunk_server_id[32] = "Invalid Chunk server ID";
char redis_ip[32];
uint16_t redis_port = 6379;

static int mode = 0;

typedef enum redis_mode {
    SINGLE_MODE = 0x01,
    CLUSTER_MODE = 0x10,
    MIX_MODE = 0x11,
} redis_mode;

/**< function call on successfully connection to redis.*/
OnRedisConnect on_redis_connect = NULL;
bool on_redis_connect_running = false;

/**< global variables used in this module.*/
static bool is_cluster = false;
static bool running = false;
static st_mutex_t redisLock = NULL;
static st_thread_t auto_disconnect = NULL;

static void
redis_client_set_mode(bool mix_enabled)
{
    bool is_cluster = false;
    bool is_single = false;
    int index = -1;

    if (1 == get_redis_servers_nb() || do_redis_servers_contain_port(6379, &index)) {
        is_single = true;
    }

    if ((is_single && 4 <= get_redis_servers_nb()) || \
			(!is_single && 3 <= get_redis_servers_nb())) {
        is_cluster = true;
    }

    mode |= is_single ? SINGLE_MODE : 0;
    mode |= is_cluster ? CLUSTER_MODE : 0;

    if (!mix_enabled && mode == MIX_MODE) {
        /**< prefer cluster */
        LOG_ERROR("[%s:%d] wrong redis configuration: use redis cluster", __func__, __LINE__);
        mode = CLUSTER_MODE;
        remove_endpoint_from_redis_servers(index);
    }

    LOG_DEBUG("[%s:%d] mix_enabled[%d] current redis mode [%d]",
              __func__, __LINE__, mix_enabled, mode);
}

static inline bool
redis_client_is_cluster(int cur_mode)
{
    if (cur_mode != -1) {
        return (cur_mode & CLUSTER_MODE) == CLUSTER_MODE;
    }

    return (mode & CLUSTER_MODE) == CLUSTER_MODE;
}

static inline bool
redis_client_is_single(int cur_mode)
{
    if (cur_mode != -1) {
        return (cur_mode & SINGLE_MODE) == SINGLE_MODE;
    }

    return (mode & SINGLE_MODE) == SINGLE_MODE;
}

#define CONNECTION_DURATION 60
static void *do_auto_disconnect(void *arg)
{
    LOG_DEBUG("start a do_auto_disconnect thread");
    st_sleep(-1);	// wait for the first connection

    while(running)
    {
        int ret = st_sleep(CONNECTION_DURATION);
        if (ret < 0 && errno == EINTR)
            continue;

        // not using blocking lock, to avoid the lock being interrupted by
        // front-end thread when redis_executev_result() or redis_client_fini()
        ret = st_mutex_trylock(redisLock);
        if (unlikely(ret < 0))
            continue;
        if (redis_client_is_cluster(-1)) {
//            redis_cluster_disconnect();
            LOG_ERROR("Redis cluster mode is not yet supported.");
            return NULL;
        }

        if (redis_client_is_single(-1)) {
            redis_single_disconnect();
        }

        st_mutex_unlock(redisLock);

        st_sleep(-1);	// wait for the next connection
    }
    LOG_DEBUG("do_auto_disconnect exit !!!");
    return NULL;
}

static int
redis_executev_result(const char *cmd, va_list arglist, int exe_mode,
        struct redisReply **reply)
{
    va_list copied;
    struct redisReply *r_single = NULL;
    struct redisReply *r_cluster = NULL;
    int ret_mode = exe_mode;
    *reply = NULL;

    if (redis_client_is_single(exe_mode)) {
        va_copy(copied, arglist);
        r_single = redis_single_executev_result(cmd, copied);
        va_end(copied);

        if (NULL != r_single) {
            ret_mode &= (~SINGLE_MODE);
        } else {
            LOG_INFO("[%s:%d] exe_mode[%d]:single redis execute CMD[%s] failed",
                     __func__, __LINE__, exe_mode, cmd);
        }

        *reply = r_single;
    }

    if (redis_client_is_cluster(exe_mode))
    {
        LOG_ERROR("redis cluster not yet supported.");
        *reply = NULL;
    }
//    {
//        va_copy(copied, arglist);
//        r_cluster = redis_cluster_executev_result(cmd, copied);
//        va_end(copied);
//
//        if (NULL != r_cluster) {
//            ret_mode &= (~CLUSTER_MODE);
//        } else {
//            LOG_INFO("[%s:%d] exe_mode[%d]: cluster redis execute CMD[%s] failed",
//                     __func__, __LINE__, exe_mode, cmd);
//        }
//
//        *reply = r_cluster;
//    }

    /**< free unused one in mix mode */
    if (NULL != r_single && *reply != r_single) {
        freeReplyObject(r_single);
    }

    if (NULL != r_cluster && *reply != r_cluster) {
        freeReplyObject(r_cluster);
    }

    if (0 != ret_mode) {
        if (NULL != *reply) {
            freeReplyObject(*reply);
        }
        *reply = NULL;
    }

    return ret_mode;
}

void
redis_conn_disconnect(int exe_mode)
{
    if (redis_client_is_single(exe_mode)) {
        redis_single_disconnect();
    }

    if (redis_client_is_cluster(exe_mode))
    {
        LOG_ERROR("redis cluster not yet supported.");
    }
//    {
//        redis_cluster_disconnect();
//    }

    LOG_DEBUG("[%s:%d] disconnect redis, exe_mode[%d]", __func__, __LINE__, exe_mode);
}

int
redis_conn_connect(int exe_mode)
{
    int ret_single = 0;
    int ret_cluster = 0;

    if (redis_client_is_single(exe_mode)) {
        ret_single = redis_single_connect();
        if (0 == ret_single) {
            LOG_DEBUG("[%s:%d] single redis connection completes successfully", __func__, __LINE__);
        }
    }

    if (redis_client_is_cluster(exe_mode))
    {
        LOG_ERROR("redis cluster not yet supported.");
        return -1;
    }
//    {
//        ret_cluster = redis_cluster_connect();
//        if (0 == ret_cluster) {
//            LOG_DEBUG("[%s:%d] single redis connection completes successfully", __func__, __LINE__);
//        }
//    }

    LOG_DEBUG("[%s:%d] redis connection: exe_mode[%d] single[%d] cluster[%d]",
              __func__, __LINE__, exe_mode, ret_single, ret_cluster);

    return (ret_single + ret_cluster);
}

static struct redisReply *
redis_executev_common(const char *cmd, va_list arglist)
{
    struct redisReply *r = NULL;
    uint32_t connect_interval = 1;
    int lockedByMe = 1;
    va_list copied;
    int exe_mode = mode;

    if (0 != st_mutex_lock(redisLock)) {
        if (EDEADLK !=  errno) {
            LOG_ERROR("[%s:%d] Cannot get redisLock. Command execution failed[%s]",
                      __func__, __LINE__, cmd);
            return NULL;
        } else {
            // EDEADLK ==  errno
            // locked twice
            LOG_INFO("redis access locked twice");
            lockedByMe = 0; // go done
        }
    }

    do {
        va_copy(copied, arglist);
        exe_mode = redis_executev_result(cmd, copied, exe_mode, &r);
        va_end(copied);

        if (0 == exe_mode) {
            st_thread_interrupt(auto_disconnect);
            LOG_INFO("[%s:%d] Command [%s] executed successfully, current redis mode[%d]",
                      __func__, __LINE__, cmd, mode);
            break;
        }

        LOG_INFO("[%s:%d] Command [%s] failed, current redis mode[%d] exe_mode[%d] TRY TO RECONNECT TO REDIS",
                 __func__, __LINE__, cmd, mode, exe_mode);
        st_usleep(connect_interval * 100 * 1000);
        redis_conn_disconnect(exe_mode);
        if (0 == redis_conn_connect(exe_mode) && on_redis_connect
           && !on_redis_connect_running) {
            on_redis_connect_running = true;
            on_redis_connect();
            on_redis_connect_running = false;
        }

        if (connect_interval < 64) {
            connect_interval *= 2;
        } else {
            break;
        }
    } while(1);

    if (lockedByMe) {
        st_mutex_unlock(redisLock);
    }

    return r;
}










struct redisReply *
redis_execute_result(const char* cmd, ...)
{
    struct redisReply *reply;
    va_list arglist;
    va_start(arglist, cmd);

    reply =  redis_executev_common(cmd, arglist);
    va_end(arglist);

    return reply;

}

int query_servers(struct CSEP *servers, int servers_len, const char* key)
{
    LOG_DEBUG("query_servers SMEMBERS %s", key);
    struct redisReply* r = redis_execute_result("SMEMBERS %s", key);
    if (r == NULL) {
        LOG_ERROR("failed to query redis for %s", key);
        return -1;
    }
    LOG_DEBUG("query_servers exeuted done");
    if (r->type != REDIS_REPLY_ARRAY) {
        LOG_ERROR("query_replicas: invalid result type %d, key %s", r->type, key);
        freeReplyObject(r);
        return -1;
    }

    uint32_t i, m;
    uint32_t n = MIN((int)r->elements, servers_len);
    for (i=m=0; i<n; ++i) {
        if (r->element[i]->type != REDIS_REPLY_STRING) {
            LOG_ERROR("query_replicas: invalid result type of element[%u] key %s", i, key);
            continue;
        }
        int ret = parse_ip_port(r->element[i]->str, &servers[m].ip, &servers[m].port, 0);
        if (ret < 0) {
            LOG_ERROR("query_replicas: invalid IP:port format of element[%u]: '%s', key %s", i, r->element[i]->str, key);
            continue;
        }
        m++;
    }

    freeReplyObject(r);
    memset(servers+m, 0, (servers_len - m) * sizeof(servers[0]));
    return m;
}

#define MANAGER_SET_NAME "Managers"
#define MANAGER_LOG_NAME "ManagerLog"

int query_managers(struct CSEP *servers, int servers_len)
{
    const char *key = MANAGER_SET_NAME;
    return query_servers(servers, servers_len, key);
}

int check_manager_in_redis()
{
    struct CSEP servers[32];
    int ret = query_managers(servers, LEN(servers)); // number of returned managers
    if(ret <= 0) {
        LOG_INFO("query managers ret %d", ret);
        return ret;
    }
    int count = ret;
    for(int i = 0; i < count; i++) {
        add_a_manager(servers[i].ip, servers[i].port);
    }
    return 0;
}

int redis_client_init(bool enable_mix)
{
//    register_local_log_level("c.redis-client");

    int ret = 0;
    redis_client_set_mode(enable_mix);

    redisLock = st_mutex_new();
    if (redisLock == NULL) {
        LOG_FATAL("[%s:%d]FAILED TO INITIALIZATION REDISLOCK.",
                  __func__, __LINE__);
        return -1;
    }

    if (redis_client_is_cluster(-1)) {
        /**< redis cluster init.*/
//        ret = redis_cluster_init();

        LOG_ERROR("Redis cluster mode is not yet supported.");
        return -1;

        /**
        redis_executev_result = redis_cluster_executev_result;
        redis_conn_disconnect = redis_cluster_disconnect;
        redis_conn_connect = redis_cluster_connect;
        */
    }

    if (redis_client_is_single(-1)) {
        ret = redis_single_init();
        /**
        redis_executev_result = redis_single_executev_result;
        redis_conn_disconnect = redis_single_disconnect;
        redis_conn_connect = redis_single_connect;
        */
    }

    if (0 != ret) {
        LOG_ERROR("FAILED TO CONNECT TO REDIS FOR NOW, WILL TRY LATER");
    }

    LOG_DEBUG("[%s:%d] REDIS CLIENT [%s] INITIALIZATION SUCCESSFULLY.",
              __func__, __LINE__, (is_cluster ? "CLUSTER" : "SINGLE"));
    running = true;
    /**< only call when succefully connected.*/
    if ((0 == ret) && on_redis_connect) {
        on_redis_connect();
    }

    auto_disconnect = st_thread_create(do_auto_disconnect, NULL, 0, 0);
    return 0;
}

int redis_execute(const char* cmd, ...)
{
    struct redisReply *reply = NULL;
    va_list arglist;

    va_start(arglist, cmd);
    reply = redis_executev_common(cmd, arglist);
    va_end(arglist);

    if (reply->type == 6 && reply->str != NULL) {
        if (strlen(reply->str) >=3 && strncmp(reply->str, "ERR", 3) == 0) {
            LOG_ERROR("error execute redis_execute: %s", reply->str);
            freeReplyObject(reply);
            return -1;
        }
    }
    if (likely(reply)) {
        freeReplyObject(reply);
        return 0;
    }
    else {
        LOG_ERROR("error execute redis_execute");
        return -1;
    }
}

int master_clear_chunks()
{
    LOG_DEBUG("enter %s", __func__);
    struct redisReply* r = redis_execute_result("SMEMBERS %s.chunks",
            chunk_server_id);
    if (r == NULL) {
        LOG_ERROR("error master_clear_chunks ret");
        return -1;
    }

    for (uint32_t i=0; i < r->elements; ++i)
    {
        uint64_t chunkid;
        int ret = sscanf(r->element[i]->str, "%llx", (long long unsigned int*)&chunkid);
        if (unlikely(ret != 1)) {
            LOG_DEBUG("error sscanf");
            continue;
        }
        master_remove_chunk(chunkid);
    }

    freeReplyObject(r);
    return 0;
}

int redis_client_fini()
{
    running = false;
    if (auto_disconnect) {
        st_thread_interrupt(auto_disconnect);
        st_sleep(0);
    }

    if (redis_client_is_cluster(-1)) {
        LOG_INFO("fini cluster redis client: cluster mode not yet supported");
//        redis_cluster_finit();
    }

    if (redis_client_is_single(-1)) {
        LOG_INFO("fini none cluster redis");
        redis_single_finit();
    }

    st_mutex_destroy(redisLock);

    return 0;
}